<!DOCTYPE html>
<html lang="zh-CN">
    <head>
        <meta charset="utf-8">
        <meta name="viewport" content="width=device-width, initial-scale=1">
        <meta name="robots" content="noodp" />
        <meta http-equiv="X-UA-Compatible" content="IE=edge, chrome=1">
        <title>《Dart异步编程之Stream》 - 飞雪无情的博客</title><meta name="Description" content="专注于IT互联网，包括但不限于Go语言(golang)、Java、Android、Python、项目管理、抖音分析、软件架构等"><meta property="og:title" content="《Dart异步编程之Stream》" />
<meta property="og:description" content="
本文主要翻译 Dart Stream 文档，了解 Stream 的用法。
" />
<meta property="og:type" content="article" />
<meta property="og:url" content="https://www.flysnow.org/posts/flutter/Dart%E5%BC%82%E6%AD%A5%E7%BC%96%E7%A8%8B%E4%B9%8BStream/" />
<meta property="article:published_time" content="2021-07-31T18:38:57+08:00" />
<meta property="article:modified_time" content="2021-07-31T18:38:57+08:00" />
<meta name="twitter:card" content="summary"/>
<meta name="twitter:title" content="《Dart异步编程之Stream》"/>
<meta name="twitter:description" content="
本文主要翻译 Dart Stream 文档，了解 Stream 的用法。
"/>
<meta name="application-name" content="飞雪无情的博客">
<meta name="apple-mobile-web-app-title" content="飞雪无情的博客"><link rel="shortcut icon" type="image/x-icon" href="/favicon.ico" />
        <link rel="icon" type="image/png" sizes="32x32" href="/favicon-32x32.png">
        <link rel="icon" type="image/png" sizes="16x16" href="/favicon-16x16.png"><link rel="apple-touch-icon" sizes="180x180" href="/apple-touch-icon.png"><link rel="manifest" href="/site.webmanifest"><link rel="canonical" href="https://www.flysnow.org/posts/flutter/Dart%E5%BC%82%E6%AD%A5%E7%BC%96%E7%A8%8B%E4%B9%8BStream/" /><link rel="prev" href="https://www.flysnow.org/posts/flutter/Flutter%E5%BC%95%E6%93%8E%E5%A5%94%E6%BA%83%E5%88%86%E6%9E%90%E6%96%B9%E6%B3%95%E7%AE%80%E4%BB%8B/" /><link rel="next" href="https://www.flysnow.org/posts/flutter/Flutter%E4%B8%8D%E8%A6%81%E6%8B%85%E5%BF%83%E5%9E%83%E5%9C%BE%E5%9B%9E%E6%94%B6%E5%99%A8/" /><link rel="stylesheet" href="/lib/normalize/normalize.min.css"><link rel="stylesheet" href="/css/style.min.css"><link rel="stylesheet" href="/lib/fontawesome-free/all.min.css"><link rel="stylesheet" href="/lib/animate/animate.min.css"><script type="application/ld+json">
    {
        "@context": "http://schema.org",
        "@type": "BlogPosting",
        "headline": "《Dart异步编程之Stream》",
        "inLanguage": "zh-CN",
        "mainEntityOfPage": {
            "@type": "WebPage",
            "@id": "https:\/\/www.flysnow.org\/posts\/flutter\/Dart%E5%BC%82%E6%AD%A5%E7%BC%96%E7%A8%8B%E4%B9%8BStream\/"
        },"genre": "posts","keywords": "flutter, dart","wordcount":  2108 ,
        "url": "https:\/\/www.flysnow.org\/posts\/flutter\/Dart%E5%BC%82%E6%AD%A5%E7%BC%96%E7%A8%8B%E4%B9%8BStream\/","datePublished": "2021-07-31T18:38:57+08:00","dateModified": "2021-07-31T18:38:57+08:00","publisher": {
            "@type": "Organization",
            "name": "飞雪无情"},"author": {
                "@type": "Person",
                "name": "飞雪无情"
            },"description": ""
    }
    </script></head>
    <body header-desktop="" header-mobile=""><script type="text/javascript">(window.localStorage && localStorage.getItem('theme') ? localStorage.getItem('theme') === 'dark' : ('' === 'auto' ? window.matchMedia('(prefers-color-scheme: dark)').matches : '' === 'dark')) && document.body.setAttribute('theme', 'dark');</script>

        <div id="mask"></div><div class="wrapper"><div class="logo-wrapper">
  <a href="/%20/" class="logo">飞雪无情的博客</a>
</div>

<nav class="site-navbar">
  <ul id="menu" class="menu">
    <li class="menu-item">
      <a class="menu-item-link" href="/tools/">工具</a>
    </li><li class="menu-item">
      <a class="menu-item-link" href="/archives/">归档</a>
    </li><li class="menu-item">
      <a class="menu-item-link" href="/about/">关于</a>
    </li>
  </ul>
</nav><main class="main">
                <div class="container"><div class="toc" id="toc-auto">
            <h2 class="toc-title">Contents</h2>
            <div class="toc-content" id="toc-content-auto"></div>
        </div><article class="page single"><h1 class="single-title animated flipInX">《Dart异步编程之Stream》</h1><div class="post-meta">
            <div class="post-meta-line"><span class="post-author"><a href="/" title="Author" rel=" author" class="author"><i class="fas fa-user-circle fa-fw"></i>飞雪无情</a></span>&nbsp;<span class="post-category">included in <a href="/categories/Flutter/"><i class="far fa-folder fa-fw"></i>Flutter</a>&nbsp;<a href="/categories/dart/"><i class="far fa-folder fa-fw"></i>dart</a></span></div>
            <div class="post-meta-line"><i class="far fa-calendar-alt fa-fw"></i>&nbsp;<time datetime="2021-07-31">2021-07-31</time>&nbsp;<i class="fas fa-pencil-alt fa-fw"></i>&nbsp;2108 words&nbsp;
                <i class="far fa-clock fa-fw"></i>&nbsp;5 minutes&nbsp;</div>
        </div><div class="featured-image"><img
        class="lazyload"
        src="/svg/loading.min.svg"
        data-src="https://luckly007.oss-cn-beijing.aliyuncs.com/img/90c6cc12-742e-4c9f-b318-b912f163b8d0.png"
        data-srcset="https://luckly007.oss-cn-beijing.aliyuncs.com/img/90c6cc12-742e-4c9f-b318-b912f163b8d0.png, https://luckly007.oss-cn-beijing.aliyuncs.com/img/90c6cc12-742e-4c9f-b318-b912f163b8d0.png 1.5x, https://luckly007.oss-cn-beijing.aliyuncs.com/img/90c6cc12-742e-4c9f-b318-b912f163b8d0.png 2x"
        data-sizes="auto"
        alt="https://luckly007.oss-cn-beijing.aliyuncs.com/img/90c6cc12-742e-4c9f-b318-b912f163b8d0.png"
        title="https://luckly007.oss-cn-beijing.aliyuncs.com/img/90c6cc12-742e-4c9f-b318-b912f163b8d0.png" /></div><div class="details toc" id="toc-static"  kept="">
                <div class="details-summary toc-title">
                    <span>Contents</span>
                    <span><i class="details-icon fas fa-angle-right"></i></span>
                </div>
                <div class="details-content toc-content" id="toc-content-static"><nav id="TableOfContents">
  <ul>
    <li><a href="#single-subscription-streams">Single subscription streams</a></li>
    <li><a href="#broadcast-streams">Broadcast streams</a></li>
  </ul>

  <ul>
    <li><a href="#transform-function">transform() function</a></li>
    <li><a href="#reading-and-decoding-a-file">Reading and decoding a file</a></li>
  </ul>
</nav></div>
            </div><div class="content" id="content"><blockquote>
<p>本文主要翻译 Dart Stream 文档，了解 Stream 的用法。</p>
</blockquote>
<p>原文见<a href="https://dart.dev/tutorials/language/streams" target="_blank" rel="noopener noreffer">这里</a>。</p>
<p>重点：</p>
<ul>
<li>Steams 提供异步的数据序列</li>
<li>数据序列可以用户生成的数据或者从文件读取的数据</li>
<li>可以使用 <strong>await for</strong> 或 Stream API 的 <code>listen()</code> 方法来处理 stream</li>
<li>Streams 提供错误处理机制</li>
<li>Streams 分为两类：单一的 subscription (single subscription) 或广播 (broadcast)</li>
</ul>
<p>Future 和 Stream 类用于 Dart 异步编程。</p>
<p>Future 表示不能立即结束的计算。普通函数返回计算结果，而异步函数返回 Future，它包含最终的计算结果。Future 会在计算完成时告知结果。</p>
<p>Stream 是异步事件序列。它有点像一个异步的 Iterable，不同之处在于，你不是自己获取下一个事件，而是由 Stream 在下一个事件准备好后通知你。</p>
<h1 id="接收-stream-事件">接收 stream 事件</h1>
<p>可以通过多种方式创建 streams，这个话题后面文章再讨论。无论哪种方式创建 stream，它们的用法都相同：用于 stream 事件上进行异步 for 循环迭代 (即 <strong>await for</strong>)，就像 Iterable 中进行普通 <strong>for 循环</strong> 迭代一样。比如，</p>
<pre><code>Future&lt;int&gt; sumStream(Stream&lt;int&gt; stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}
</code></pre><p>这段代码只是接收整数事件 stream 的每个事件，然后返回它们的和 (一个 Future)。循环体结束后，函数暂停并等待下一个事件到来，直到 stream 结束。</p>
<p>使用 <strong>await for</strong> 循环时要求函数带 <code>async</code> 关键字。</p>
<h1 id="错误事件">错误事件</h1>
<p>Streams 在没有更多事件时结束。使用 <strong>await for</strong> 循环读取事件时，循环在 stream 结束时停止。</p>
<p>某些情况下，stream 结束前出现错误，比如，从远程服务器获取文件时网络断开，或者创建事件的代码有 bug。</p>
<p>Streams 以发送数据事件的相同方式发送错误事件。大多数 streams 会在第一次错误时中止，但也有些 streams 可能多次发送错误事件，还有些 streams 在错误事件后继续发送数据事件。这个文档中我们只讨论最多发送一次错误事件的情况。</p>
<p>使用 <strong>await for</strong> 读取 stream 时，循环语句抛出错误同时结束循环。可以使用 <strong>try-catch</strong> 来捕获错误。</p>
<h1 id="使用-stream">使用 stream</h1>
<p><code>Stream</code> 类有许多工具方法用于在 stream 上执行常用操作，跟 <code>Iterable</code> 的方法类似。比如，可以使用 Stream API 的 <code>lastWhere()</code> 方法查找 stream 中最后一个正整数：</p>
<pre><code>Future&lt;int&gt; lastPositive(Stream&lt;int&gt; stream) =&gt;
    stream.lastWhere((x) =&gt; x &gt;= 0);
</code></pre><h1 id="两种-stream">两种 Stream</h1>
<p>有两种 stream。</p>
<h2 id="single-subscription-streams">Single subscription streams</h2>
<p>最常用的 stream 包含整个事件中的一个小序列。事件必须以正确的顺序分发。读取文件或接收网络响应时得到的正是这种 stream。</p>
<h2 id="broadcast-streams">Broadcast streams</h2>
<p>另一种 stream 专门用于每次只能处理一条消息的场景。比如，用于浏览器中的鼠标事件。</p>
<p>可以在任何时候监听这个 stream，监听后会收到发送出来的事件。同一时刻可以监听多个 listener，取消前一个订阅后还能再次监听。</p>
<h1 id="处理-stream">处理 stream</h1>
<p><code>Stream&lt;T&gt;</code> 上的以下方法用于处理 stream 并且返回结果：</p>
<pre><code>Future&lt;T&gt; get first;
Future&lt;bool&gt; get isEmpty;
Future&lt;T&gt; get last;
Future&lt;int&gt; get length;
Future&lt;T&gt; get single;
Future&lt;bool&gt; any(bool Function(T element) test);
Future&lt;bool&gt; contains(Object needle);
Future&lt;E&gt; drain&lt;E&gt;([E futureValue]);
Future&lt;T&gt; elementAt(int index);
Future&lt;bool&gt; every(bool Function(T element) test);
Future&lt;T&gt; firstWhere(bool Function(T element) test, {T Function() orElse});
Future&lt;S&gt; fold&lt;S&gt;(S initialValue, S Function(S previous, T element) combine);
Future forEach(void Function(T element) action);
Future&lt;String&gt; join([String separator = &quot;&quot;]);
Future&lt;T&gt; lastWhere(bool Function(T element) test, {T Function() orElse});
Future pipe(StreamConsumer&lt;T&gt; streamConsumer);
Future&lt;T&gt; reduce(T Function(T previous, T element) combine);
Future&lt;T&gt; singleWhere(bool Function(T element) test, {T Function() orElse});
Future&lt;List&lt;T&gt;&gt; toList();
Future&lt;Set&lt;T&gt;&gt; toSet();
</code></pre><p>除了 <code>drain()</code> 和 <code>pipe()</code>，其他方法都能在 Iterable 中找到相应的方法。这些方法很容易写，在 async 方法使用 <strong>await for</strong> 循环。比如，以下是一些方法的实现：</p>
<pre><code>Future&lt;bool&gt; contains(Object needle) async {
  await for (var event in this) {
    if (event == needle) return true;
  }
  return false;
}

Future forEach(void Function(T element) action) async {
  await for (var event in this) {
    action(event);
  }
}

Future&lt;List&lt;T&gt;&gt; toList() async {
  final result = &lt;T&gt;[];
  await this.forEach(result.add);
  return result;
}

Future&lt;String&gt; join([String separator = &quot;&quot;]) async =&gt;
    (await this.toList()).join(separator);
</code></pre><p>(注意，以上只是精简后的代码，由于历史原因实际实现可能更复杂。)</p>
<h1 id="修改-stream">修改 stream</h1>
<p>Stream 类的以下方法基于原先的 stream 返回一个新的 stream。Each one waits until someone listens on the new stream before listening on the original.</p>
<pre><code>Stream&lt;R&gt; cast&lt;R&gt;();
Stream&lt;S&gt; expand&lt;S&gt;(Iterable&lt;S&gt; Function(T element) convert);
Stream&lt;S&gt; map&lt;S&gt;(S Function(T event) convert);
Stream&lt;R&gt; retype&lt;R&gt;();
Stream&lt;T&gt; skip(int count);
Stream&lt;T&gt; skipWhile(bool Function(T element) test);
Stream&lt;T&gt; take(int count);
Stream&lt;T&gt; takeWhile(bool Function(T element) test);
Stream&lt;T&gt; where(bool Function(T event) test);
</code></pre><p>相应地，Iterable 中也有类似的方法用于将一个 iterable 转换成另外一个 iterable。这些方法很容易使用带 <strong>await for</strong> 循环的 async 方法编写。</p>
<pre><code>Stream&lt;E&gt; asyncExpand&lt;E&gt;(Stream&lt;E&gt; Function(T event) convert);
Stream&lt;E&gt; asyncMap&lt;E&gt;(FutureOr&lt;E&gt; Function(T event) convert);
Stream&lt;T&gt; distinct([bool Function(T previous, T next) equals]);
</code></pre><p><code>asyncExpand()</code> 和 <code>asyncMap()</code> 函数跟 <code>expand()</code> 和 <code>map()</code> 类似，但允许其函数参数为异步函数。</p>
<pre><code>Stream&lt;T&gt; handleError(Function onError, {bool test(error)});
Stream&lt;T&gt; timeout(Duration timeLimit,
    {void Function(EventSink&lt;T&gt; sink) onTimeout});
Stream&lt;S&gt; transform&lt;S&gt;(StreamTransformer&lt;T, S&gt; streamTransformer);
</code></pre><p>最后的三个方法更特别。它们用于处理 <strong>await for</strong> 循环不能处理的错误，循环中首次出现的错误会导致循环结束，对 stream 的订阅也被结束。没法从这种情况恢复。可以使用 <code>handleError()</code> 在错误事件到达 <strong>await for</strong> 循环之前从 stream 中移除错误。(原文：You can use handleError() to remove errors from a stream before using it in an await for loop.)</p>
<h2 id="transform-function">transform() function</h2>
<p><code>transform()</code> 函数不仅仅用于错误处理，还能用于对 stream 进行更通用的 “map”。通常的 map 要求为每个收到的事件得到一个值。但是在 I/O 操作中，常常出现几个输入事件生成一个输出事件的情形。<code>StreamTransformer</code> 用于处理这种情况。比如，<code>Utf8Decoder</code> 等解码器也是 transformer。transformer 只需要实现 <code>bind()</code> 方法。</p>
<pre><code>Stream&lt;S&gt; mapLogErrors&lt;S, T&gt;(
  Stream&lt;T&gt; stream,
  S Function(T event) convert,
) async* {
  var streamWithoutErrors = stream.handleError((e) =&gt; log(e));
  await for (var event in streamWithoutErrors) {
    yield convert(event);
  }
}
</code></pre><h2 id="reading-and-decoding-a-file">Reading and decoding a file</h2>
<p>以下代码读文件并且对 stream 运用两个 transform。首先从 UTF8 转换数据，然后再使用 <code>LineSplitter</code>。最后打印所有行，除了以 <code>#</code> 开头的行。</p>
<pre><code>import 'dart:convert';
import 'dart:io';

Future&lt;void&gt; main(List&lt;String&gt; args) async {
  var file = File(args[0]);
  var lines = utf8.decoder
      .bind(file.openRead())
      .transform(LineSplitter());
  await for (var line in lines) {
    if (!line.startsWith('#')) print(line);
  }
}
</code></pre><h1 id="the-listen-method">The listen() method</h1>
<p>最后要介绍的是 Stream 中的 <code>listen()</code> 方法。它是一个低级方法，所有的其他 stream 方法都是某种意义上的 <code>listen()</code> 方法。</p>
<pre><code>StreamSubscription&lt;T&gt; listen(void Function(T event) onData,
    {Function onError, void Function() onDone, bool cancelOnError});
</code></pre><p>要创建新的 <code>Stream</code> 类型，只需要继承 <code>Stream</code> 类并且实现 <code>listen()</code> 方法。<code>Stream</code> 上的其他方法都会调用 <code>listen()</code> 方法。</p>
<p><code>listen()</code> 方法允许监听 stream。直接被监听，否则 stream 就是静态的。开始监听后，将返回 <code>StreamSubscription</code> 对象，它代表活动的、可以产生事件的流。这跟 <code>Iterable</code> 类似，它只是对象集合，itertor 才会真正进行迭代。</p>
<p>stream subscription 允许暂停以及恢复，或者完全取消。还能设置回调，当产生数据事件、错误事件或者 stream 被关闭时都会收到回调。</p>
<h1 id="其他资源">其他资源</h1>
<p>阅读以下 stream 用法及 Dart 异步编程文档，了解更多内容。</p>
<ul>
<li><a href="https://dart.dev/articles/libraries/creating-streams" target="_blank" rel="noopener noreffer">Creating Streams in Dart</a>, an article about creating your own streams</li>
<li><a href="https://dart.dev/guides/libraries/futures-error-handling" target="_blank" rel="noopener noreffer">Futures and Error Handling</a>, an article that explains how to handle errors using the Future API</li>
<li><a href="https://dart.dev/guides/language/language-tour#asynchrony-support" target="_blank" rel="noopener noreffer">Asynchrony support</a>, a section in the <a href="https://dart.dev/guides/language/language-tour" target="_blank" rel="noopener noreffer">language tour</a></li>
<li><a href="https://api.dart.dev/stable/dart-async/Stream-class.html" target="_blank" rel="noopener noreffer">Stream API reference</a></li>
</ul></div><div class="post-footer" id="post-footer">
    <div class="post-info">
        <div class="post-info-line">
            <div class="post-info-mod">
                <span>Updated on 2021-07-31</span>
            </div>
            <div class="post-info-license"></div>
        </div>
        <div class="post-info-line">
            <div class="post-info-md"></div>
            <div class="post-info-share">
                <span></span>
            </div>
        </div>
    </div>

    <div class="post-info-more">
        <section class="post-tags"><i class="fas fa-tags fa-fw"></i>&nbsp;<a href="/tags/flutter/">flutter</a>,&nbsp;<a href="/tags/dart/">dart</a></section>
        <section>
            <span><a href="javascript:void(0);" onclick="window.history.back();">Back</a></span>&nbsp;|&nbsp;<span><a href="/">Home</a></span>
        </section>
    </div>

    <div class="post-nav"><a href="/posts/flutter/Flutter%E5%BC%95%E6%93%8E%E5%A5%94%E6%BA%83%E5%88%86%E6%9E%90%E6%96%B9%E6%B3%95%E7%AE%80%E4%BB%8B/" class="prev" rel="prev" title="《Flutter引擎奔溃分析方法简介》"><i class="fas fa-angle-left fa-fw"></i>《Flutter引擎奔溃分析方法简介》</a>
            <a href="/posts/flutter/Flutter%E4%B8%8D%E8%A6%81%E6%8B%85%E5%BF%83%E5%9E%83%E5%9C%BE%E5%9B%9E%E6%94%B6%E5%99%A8/" class="next" rel="next" title="《Flutter不要担心垃圾回收器》">《Flutter不要担心垃圾回收器》<i class="fas fa-angle-right fa-fw"></i></a></div>
</div>
</article></div>
            </main><footer class="footer">
        <div class="footer-container"><div class="footer-line">Powered by <a href="https://gohugo.io/" target="_blank" rel="noopener noreffer" title="Hugo 0.79.1">Hugo</a> | Theme - <a href="https://github.com/dillonzq/LoveIt" target="_blank" rel="noopener noreffer" title="LoveIt 0.2.10"><i class="far fa-kiss-wink-heart fa-fw"></i> LoveIt</a>
                </div><div class="footer-line"><i class="far fa-copyright fa-fw"></i><span itemprop="copyrightYear">2021</span><span class="author" itemprop="copyrightHolder">&nbsp;<a href="/" target="_blank">飞雪无情</a></span></div>
        </div>
    </footer></div>

        <div id="fixed-buttons"><a href="#" id="back-to-top" class="fixed-button" title="Back to Top">
                <i class="fas fa-arrow-up fa-fw"></i>
            </a><a href="#" id="view-comments" class="fixed-button" title="View Comments">
                <i class="fas fa-comment fa-fw"></i>
            </a>
        </div><script type="text/javascript" src="/lib/smooth-scroll/smooth-scroll.min.js"></script><script type="text/javascript" src="/lib/lazysizes/lazysizes.min.js"></script><script type="text/javascript" src="/lib/clipboard/clipboard.min.js"></script><script type="text/javascript" src="/lib/sharer/sharer.min.js"></script><script type="text/javascript">window.config={"code":{"copyTitle":"Copy to clipboard","maxShownLines":10},"comment":{}};</script><script type="text/javascript" src="/js/theme.min.js"></script></body>
</html>
